package com.juneyao.bigdata.service.impl;

import com.juneyao.bigdata.common.ConstantNV;
import com.juneyao.bigdata.domain.*;
import com.juneyao.bigdata.service.ActionService;
import com.juneyao.bigdata.service.ElasticSearchService;
import com.juneyao.bigdata.service.HbaseSearchService;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author quanbiao
 * @date 2018-06-29
 * 活动发送实现类
 */
@Service("actionService")
public class ActionServiceImpl implements ActionService{

    private Logger logger = LoggerFactory.getLogger("ActionServiceImpl");

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Autowired
    private ElasticSearchService elasticSearchService;

    @Autowired
    private HbaseSearchService hbaseSearchService;

    /**
     * 0标识活动通知
     */
    private static final String PRODUCTION_NUM_ZERO = "0";

    /**
     * 短信标识，1代表有短信
     */
    private static final String MSM_FLAG = "1";

    /**
     * APP标识符
     */
    private static final String APP_FLAG = "3";

    /**
     * 核心线程
     */
    private static final int CORE_POOL_SIZE = 10;

    /**
     * 最大线程
     */
    private static final int MAX_NUM_POOL_SIZE = 20;

    /**
     * 线程空闲存活时间
     */
    private static final long THREAD_POOL_KEEP_ALIVE_TIME = 10000L;

    /**
     * 线程池
     */
    private static final ThreadPoolExecutor threadPool;

    private final Object lock = new Object();

    @Value("${jx.user.inter.topic}")
    private String kafkaTopic;

    /**
     * 初始化线程池
     */
    static {
        threadPool = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_NUM_POOL_SIZE,
                THREAD_POOL_KEEP_ALIVE_TIME,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>());
        threadPool.allowCoreThreadTimeOut(true);
    }

    /**
     * 活动审核后-活动确认发送
     * @param act 活动对象
     * @return msg
     */
    @Override
    public ResultFlag activitySend(Activity act){

        threadPool.submit(
                () -> this.activitySendThread(act)
        );
        ResultFlag rf = new ResultFlag();
        rf.setErrorCode(0);
        rf.setMsg("receive success!");
        return rf;
    }

    /**
     * 活动发送-异步
     * @param act
     */
    public void activitySendThread(Activity act){
        //集合str
        String peopleGroupIdStr = act.getPeopleGroupId();

        //活动父id
        String parentActId = act.getParentActId();
        //渠道
        String channel = act.getChannel();
        //产品数量
        int productNum = Integer.valueOf(act.getProduction_num());

        //该值为0时，本活动为通知发送，二次过滤不需要考虑
        if(StringUtils.equals(PRODUCTION_NUM_ZERO, act.getProduction_num())) {
            //查询build
            BoolQueryBuilder boolQueryBuilder = elasticSearchService.getBoolQueryBuilder(peopleGroupIdStr, ConstantNV.MEMBER_FLAG_YES, ConstantNV.BLOCK_FLAG_YES, ConstantNV.WEI_CHAT_FLAG_YES);
            //获取总的记录数，用于分批次查询
            SearchQuery searchQuery = new NativeSearchQueryBuilder()
                    //索引名称
                    .withIndices(ConstantNV.ES_INDEX_NAME)
                    //索引类型
                    .withTypes(ConstantNV.ES_INDEX_TYPE)
                    .withFilter(boolQueryBuilder).build();
            long userTotalCount = elasticSearchService.searchCount(searchQuery);
            //每页条数-考虑到短信成本，最多发送10万条
            int pageSize = 100000;
            //随机第几页，用于发送的人员
            int orderToSendPageNum = 0;
            long totalPages = userTotalCount % pageSize==0? userTotalCount/pageSize : (userTotalCount/pageSize)+1;
            if(totalPages > 0){
                Random random = new Random();
                orderToSendPageNum = random.nextInt((int)totalPages);
            }
            //用户id集合
            List<String> userIds = elasticSearchService.getUserIdsForAction(boolQueryBuilder, orderToSendPageNum, pageSize);
            if(userIds.size() == 0){
                logger.info("==会员标签过滤后数据为空！");
            }else{
                //微信和app都带会员id，短信带手机号
                detailAssembleAndSendKafka(userIds,act);
            }

        }else {
            //该值非0时，本活动为产品发送，需要考虑二次过滤
            //加锁
            synchronized (lock){
                //1.首先根据活动父id，去hbase中查询
                Result retByRowKeyColumn = hbaseSearchService.getResultByrowkeyColumn("tbl_act_pid_user_ids", parentActId, "cf1", "userIds");

                //如果不为空，说明hbase已经有值
                if(!retByRowKeyColumn.isEmpty()){
                    List<String> alreadyUserIds = new ArrayList<>();
                    Cell[] cells = retByRowKeyColumn.rawCells();
                    for (Cell cell : cells) {
                        String[] aUserIds = new String(CellUtil.cloneValue(cell)).split(",",-1);
                        Collections.addAll(alreadyUserIds, aUserIds);
                    }
                    //防止超发，获取等于产品数量的人员进行发送
                    List<String> retUserIds = arrSizeComp(alreadyUserIds, act);
                    //直接发送
                    detailAssembleAndSendKafka(retUserIds, act);
                }else{
                    //如果hbase中没有数据，首先判断是否有短信标识 isSms 是否有短信 1：0 （1：有，0：没有）
                    String isSmsFlag = act.getIsSms();
                    //需要发送的用户id
                    List<String> sendUserIds;
                    BoolQueryBuilder boolQueryBuilder;
                    //如果是有短信标识或者渠道是app（因为短信和app的判断是一致的）
                    if(StringUtils.equals(MSM_FLAG, isSmsFlag) || StringUtils.equals(channel, APP_FLAG)){
                        boolQueryBuilder = elasticSearchService.getBoolQueryBuilder(peopleGroupIdStr,
                                ConstantNV.MEMBER_FLAG_YES, ConstantNV.BLOCK_FLAG_YES, ConstantNV.WEI_CHAT_FLAG_NO);
                    }else{
                        //如果没短信标识,渠道也不是app，那说明是微信
                        //微信id集合
                        boolQueryBuilder = elasticSearchService.getBoolQueryBuilder(peopleGroupIdStr,
                                ConstantNV.MEMBER_FLAG_NO, ConstantNV.BLOCK_FLAG_YES, ConstantNV.WEI_CHAT_FLAG_YES);
                    }
                    sendUserIds = elasticSearchService.getUserIdsForAction(boolQueryBuilder, 0,productNum);
                    //取小于产品数量的用户id
                    List<String> retSendUserIds = arrSizeComp(sendUserIds, act);
                    //逗号拼接
                    String userIdsCompStr = StringUtils.join(retSendUserIds, ",");
                    //插入到hbase
                    hbaseSearchService.insertOneRow(parentActId,userIdsCompStr);
                    //发送--防止超发
                    detailAssembleAndSendKafka(sendUserIds, act);
                }
            }
        }
    }

    /**
     * 随机抽取
     * @param targetUserIds
     * @param act
     */
    private List<String> arrSizeComp(List<String> targetUserIds, Activity act){
        if(targetUserIds.size() <= Integer.valueOf(act.getProduction_num())){
            return targetUserIds;
        }else{
            Integer productionNum = Integer.valueOf(act.getProduction_num());

            List<String> finalUserIdArr = new ArrayList<>(targetUserIds);
            //随机抽取
            Set<String> randomUserIds =  new HashSet<>();
            for (int i = 0; i < finalUserIdArr.size(); i++) {
                int index=(int)(Math.random()*(targetUserIds.size()));
                randomUserIds.add(finalUserIdArr.get(index));
                if(randomUserIds.size() == productionNum){
                    break;
                }
            }
            List<String> retUserIds = new ArrayList<>();
            retUserIds.addAll(randomUserIds);
            return retUserIds;
        }

    }

    /**
     * 获取会员信息，并发送kafka消息
     * @param userIds
     * @param act
     */
    public void detailAssembleAndSendKafka(List<String> userIds, Activity act) {

        //用户详情
        List<UserSendVo> userSendDetail = hbaseSearchService.getUserSendDetail(ConstantNV.HBASE_USER_DETAIL,
                userIds, act.getChannel());
        //kafka发送
        for (UserSendVo userSendVo : userSendDetail) {
            //存储在hbase里面的用户id格式：xx_id,使用_分割，取第二个
//            String[] splitUserId = userSendVo.getUserId().split("_");
//            String userId = splitUserId[1];
            //kafka发送
            kafkaSending(act, userSendVo.getUserId(), userSendVo);
        }
    }

    /**
     * kafka发送
     * @param act
     * @param pid
     */
    public void kafkaSending(Activity act, String pid, UserSendVo userSendVo) {
        ActionSend as = new ActionSend();
        as.setActId(act.getActId());
        as.setPeopleid(pid);
        as.setChannel(act.getChannel());
        as.setMemberSend(userSendVo.getMobile_phone());
        as.setMemberId(userSendVo.getUserId());
        as.setMemberCode(userSendVo.getPk_member_id());
        logger.info("kafka sending message="+as.toString());
        //kafka发送
        kafkaTemplate.send(kafkaTopic, userSendVo.getPk_crm_info(), as.toString());
    }

    /**
     * 产品使用量
     * @param peopleGroupId
     * @return
     */
    @Override
    public List<Activity> getActivityUsed(PeopleGroupId peopleGroupId) {
        String[] ActIDS = peopleGroupId.getIds().split(",",-1);
        List<Activity> activities = new ArrayList<>();
        Activity activity1 = new Activity();
        for (int i=0; i<ActIDS.length; i++) {
            activity1.setActId(ActIDS[i]);
            Result r1 = hbaseSearchService.getResultByrowkeyColumn("tbl_pm_activity_cnt",ActIDS[i],"cf1","name");
            Cell[] cells = r1.rawCells();
            for (Cell cell : cells) {
                activity1.setActName(new String(CellUtil.cloneValue(cell)));
            }

            Result r2 = hbaseSearchService.getResultByrowkeyColumn("tbl_pm_activity_cnt",ActIDS[i],"cf1","count");
            Cell[] cells2 = r2.rawCells();
            for (Cell cell : cells2) {
                activity1.setProduction_num(new String(CellUtil.cloneValue(cell)));
            }
            activities.add(activity1);
        }
        return activities;

    }


    public static void main(String[] args){
        int pageSize = 5;
        int userTotalCount = 26;
        long totalPages = userTotalCount % pageSize==0? userTotalCount/pageSize : (userTotalCount/pageSize)+1;

        Random ran1 = new Random();
        System.out.println("使用种子为10的Random对象生成[0,10)内随机整数序列: ");
        for (int i = 0; i < 1000; i++) {
            System.out.print(ran1.nextInt(10) + " ");
        }

    }
}
